import com.google.common.base.Strings;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class DiskToDfs {
    public static void main(String[] args) throws IOException {
        String temp = args[0];
        DiskTo(temp);

        String path ="/tmp/dfs.txt";
        ReadFi(path);

    }

    public static void DiskTo(String path)throws IOException{
        //1.创建词汇表Configuration类型的对象
        Configuration conf =  new Configuration();
        conf.set("fs.defaultFS","hdfs://20210322045-master:9000");

        //2获取输入的本地文件路径args[0]

        String temp = path;
        FileSystem fs = FileSystem.get(conf);
        FileInputStream in1 = new FileInputStream(temp);
        System.out.println("你输入的本地文件位置为："+temp);
        FSDataOutputStream out1 = fs.create(new Path("/tmp/dfs.txt"));
        IOUtils.copyBytes(in1,out1,conf,false);
        System.out.println("本地文件上传hdfs成功，上传的文件存放在hdfs的/tmp/dfs.txt里面");
        out1.close();
        in1.close();
        fs.close();

    }

    public static void ReadFi(String path) throws IOException{
//        if(path.length == 0){
//            System.out.println("创建失败，请传入一个路径参数指定要读取的文件");
//            return;
//        }
        String feilePath = path;
        System.out.println("传入的名字是:"+feilePath);

        //1.创建词汇表Configuration类型的对象
        Configuration conf =  new Configuration();
        conf.set("fs.defaultFS","hdfs://20210322045-master:9000");

        FileSystem fs = FileSystem.get(conf);
        FSDataInputStream fsDataInputStream = fs.open(new Path(feilePath));
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream));

        //读取文件第一行字符串，如果返回null，表明已经读取到文件的末尾
        String nextLine = bufferedReader.readLine();
        HashMap<String,Integer> wordCs = new HashMap<String, Integer>();//用来统计单词的次数

        //如果没有读取到末尾，则继续读取
        while (null != nextLine){
            String []words=nextLine.split(" ");//分割字符串
            int i=1;
            for(String word:words){//遍历字符串数组
                word=word.toLowerCase();//将单词转换成小写
                //判断集合中是否已存有不区分大小写的相同字符串，同时判断字符串是否是空串
                if(!wordCs.containsKey(word)&&!word.equals(""))
                {//将字符串存储到集合中
                    System.out.println("出现的单词有："+word);
                    wordCs.put(word,i);
                }
                else {
                    i++;
                    wordCs.put(word,i);
                }
            }

            System.out.println("此句子共出现"+wordCs.size()+"个单词。");
            System.out.println("每个单词的个数为："+wordCs);
            //将读取到的数据输出到控制台

            nextLine = bufferedReader.readLine();
        }
        fs.close();


        }

    }




